import {
  getOperator,
  Options,
  OpType,
  PipelineOperator,
  ProcessingMode,
  initOptions,
  ComputeOptions,
  Context
} from "../core";
import { AsyncIterator, AsyncLazy, AsyncSource } from "./lazy";
import { Any, AnyObject } from "../types";
import { assert, cloneDeep } from "../util";
import context from "../init/context"
/**
 * The `AsyncAggregator` class provides functionality to process data collections
 * through an aggregation pipeline asynchronously. It supports streaming and executing
 * aggregation operations with customizable options.
 */
export class AsyncAggregator {
  #pipeline: AnyObject[];
  #options: Options;

  /**
   * Creates an instance of the AsyncAggregator class.
   *
   * @param pipeline - An array of objects representing the aggregation pipeline stages.
   * @param options - Optional configuration settings for the aggregator.
   */
  constructor(pipeline: AnyObject[], options?: Partial<Options>) {
    this.#pipeline = pipeline;
    const opts = options instanceof ComputeOptions ? options.getOptions() : { ...options };
    (opts as any).isAsync = true;
    this.#options = initOptions({...opts,context: context()});
  }

  /**
   * Processes a collection through an aggregation pipeline and returns an async iterator
   * for the transformed results.
   *
   * @param collection - The input collection to process. This can be any source
   *   that implements the `AsyncSource` interface.
   * @param options - Optional configuration for processing. If not provided, the
   *   default options of the aggregator instance will be used.
   * @returns An async iterator that yields the results of the aggregation pipeline.
   *
   * @throws Will throw an error if:
   * - A pipeline stage contains more than one operator.
   * - The `$documents` operator is not the first stage in the pipeline.
   * - An unregistered pipeline operator is encountered.
   */
  async  stream(collection: AsyncSource, options?: Options): Promise<AsyncIterator> {
    // Create initial async iterator from the collection
    let iter: AsyncIterator = AsyncLazy(collection);
    const opts = options ?? this.#options;
    const mode = opts.processingMode;

    // Clone the input collection if requested
    if (mode & ProcessingMode.CLONE_INPUT) {
      iter = iter.map(cloneDeep);
    }

    // Process each stage of the pipeline
    for (let i = 0; i < this.#pipeline.length; i++) {
      const stage = this.#pipeline[i];
      const keys = Object.keys(stage);

      assert(
        keys.length === 1,
        `aggregation stage must have single operator, got ${keys.toString()}.`
      );

      const name = keys[0];

      // $documents operator must be first in the pipeline
      assert(
        name !== "$documents" || i == 0,
        "$documents must be first stage in pipeline."
      );

      const op = opts.context.getOperator(OpType.PIPELINE, name) as PipelineOperator;
      assert(!!op, `unregistered pipeline operator ${name}.`);

      // Apply the operator to the iterator
      iter = (await op(iter, stage[name], opts)) as AsyncIterator;
    }

    // Clone output if requested
    if (mode & ProcessingMode.CLONE_OUTPUT) {
      iter = iter.map(cloneDeep);
    }

    return iter;
  }

  /**
   * Executes the aggregation pipeline on the provided collection and returns a Promise
   * that resolves to the resulting array.
   *
   * @template T - The type of the objects in the resulting array.
   * @param collection - The input data source to run the aggregation on.
   * @param options - Optional settings to customize the aggregation behavior.
   * @returns A Promise that resolves to an array of objects of type `T` resulting from the aggregation.
   */
  async run<T extends AnyObject>(collection: AsyncSource, options?: Options): Promise<T[]> {
    const iter = await this.stream(collection, options);


    assert(iter instanceof AsyncIterator, "AsyncIterator expected")
    return iter.value();
  }
}